package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/2/6
 *
 *      Catalog: 元数据目录。
 *
 *      假设读取mysql和oracle中的数据。
 *          但是mysql和oracle有同名的库和表，如何区分?
 *              需要在mysql和oracle的库上添加一层目录，进行区分
 *                  mysql:  mysql.db1.t1
 *                  oracle: oracle.db1.t1
 */
public class Demo6_Catelog
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = "create table t1( id string,ts bigint ,vc int ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/func/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        //执行建表(写操作)
        tableEnvironment.executeSql(createTableSql);

        //default_catalog
        System.out.println(tableEnvironment.getCurrentCatalog());
        //default_database
        System.out.println(tableEnvironment.getCurrentDatabase());

        //把t1变为Table对象
        Table table = tableEnvironment.from("t1");

        tableEnvironment.sqlQuery("select * from default_catalog.default_database.t1")
                        .execute().print();

    }
}
